package com.bkhech.netty.utils;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import com.bkhech.pojo.netty.NettyServerNode;
import com.bkhech.utils.JsonUtils;

import java.net.InetAddress;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Objects;

import static com.bkhech.ZkConstant.NODE_NAME;


/**
 * ZKUtil
 *
 * @author guowm
 * @date 2024-11-21
 */
@Slf4j
public class ZKUtil {

    /**
     * 注册当前服务到 zookeeper 节点
     * @param nodeName 服务命名空间
     * @param ip
     * @param port
     * @throws Exception
     */
    public static void registerNettyServer(String nodeName, String ip, int port) throws Exception {
        CuratorFramework client = CuratorConfig.getClient();
        String path = nodeName;
        Stat stat = client.checkExists().forPath(path);
        if (stat == null) {
            // 为空就创建节点
            client.create()
                    // 递归创建父节点，如果父节点不存在的话
                    .creatingParentsIfNeeded()
                    // 创建持久节点，因为是根节点
                    .withMode(CreateMode.PERSISTENT)
                    .forPath(path);
        } else {
            log.info("stat:{}", stat);
        }
        // 创建对应的临时有序节点，值放在线人数，初始化默认 0
        NettyServerNode nettyServerNode = new NettyServerNode();
        nettyServerNode.setIp(ip);
        nettyServerNode.setPort(port);
        String nodeJson = JsonUtils.objectToJson(nettyServerNode);
        client.create()
                // 创建临时有序节点
                .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                .forPath(path + "/im-", StrUtil.bytes(nodeJson, StandardCharsets.UTF_8));
        log.info("ip为 {}，端口号为 {} 的 chat-server 的服务已经注册到 zookeeper", ip, port);
    }

    public static String getLocalIp() throws Exception {
        return InetAddress.getLocalHost().getHostAddress();
    }

    /**
     * 在线人数 +1
     * @param serverNode
     */
    public static void incrementOnlineCount(NettyServerNode serverNode) {
        dealOnlineCount(serverNode, 1);
    }

    /**
     * 在线人数 -1
     * @param serverNode
     */
    public static void decrementOnlineCount(NettyServerNode serverNode) {
        dealOnlineCount(serverNode, -1);
    }


    /**
     * 处理在线人数增减 count
     *
     * @param serverNode
     * @param count
     */
    public static void dealOnlineCount(NettyServerNode serverNode, int count) {
        CuratorFramework client = CuratorConfig.getClient();
        // 创建读写锁（分布式锁：让当前线程排队执行，保护共享资源的安全操作）
        InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, "/rw-lock");

        final String parentPath = NODE_NAME;
        try {
            // 获取写锁
            readWriteLock.writeLock().acquire();
            List<String> nodeChildren = client.getChildren().forPath(parentPath);
            if (CollectionUtil.isEmpty(nodeChildren)) {
                return;
            }
            for (String node : nodeChildren) {
                String nodePath = parentPath + StrUtil.SLASH + node;
                String dataJson = new String(client.getData().forPath(nodePath));
                NettyServerNode pendingNode = JsonUtils.jsonToPojo(dataJson, NettyServerNode.class);
                if (pendingNode != null && Objects.equals(serverNode.getIp(), pendingNode.getIp())
                        && Objects.equals(pendingNode.getPort(), serverNode.getPort())) {
                    int onlineCounts = pendingNode.getOnlineCounts() + count;
                    if (onlineCounts < 0) {
                        pendingNode.setOnlineCounts(0);
                    } else {
                        pendingNode.setOnlineCounts(onlineCounts);
                    }

                    String newNodeJson = JsonUtils.objectToJson(pendingNode);
                    assert newNodeJson != null;
                    client.setData().forPath(nodePath, newNodeJson.getBytes(StandardCharsets.UTF_8));
                }
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            try {
                // 释放写锁
                readWriteLock.writeLock().release();
            } catch (Exception e) {
                log.info("释放所资源异常:{},{}", e, e.getMessage());
            }
        }
    }
}
